
#include <unistd.h>

#include "../iface/APRInterface.h"
#include "../iface/debug.h"
#include <apr_atomic.h>
#include <apr_file_io.h>
#include "TCPServer.h"
#include <new>

#define CONNECTION_TIMEOUT  5000
#define DATA_TIMEOUT        20000
#define POLL_TIMEOUT_USEC   5000

/** Constructor for the TCP server class.
 * 
 * @param inoofthreads The number of threads to use for serving
 * connections.
 * 
 * @param iportno The port to listen on.
 * 
 * @param ihostname The host (interface) to listen on.
 * 
 * @param upperpool The memory pool to derive this server memory pool
 * from. */
TCPServer::TCPServer(int inoofthreads, int iportno, const char *ihostname, apr_pool_t *upperpool)
{
   apr_status_t status;
   status = apr_pool_create(&mp, upperpool);
   noofthreads = inoofthreads;
   listenstop = 0;
   sessionstop = 0;
   portno = iportno;
   listenstop = 1;
   if (ihostname)
   {
      hostname = (char *)apr_palloc(mp, sizeof(char)*(strlen(ihostname)+1));
      strcpy(hostname, ihostname);
   } else hostname = NULL;
   pipearray = (ClientPipeStreams *)apr_palloc(mp, sizeof(ClientPipeStreams)*noofthreads);
   for (int i = 0; i < noofthreads; i++)
   {
      pipearray[i].recv_to_handler = NULL;
      pipearray[i].handler_to_send = NULL;
      void *mem_for_new_waiting_lock = apr_palloc(mp, sizeof(APRMutex));
      pipearray[i].waiting = new(mem_for_new_waiting_lock) APRMutex(mp);
      void *mem_for_new_sessionstart_lock = apr_palloc(mp, sizeof(APRMutex));
      pipearray[i].sessionstart = new(mem_for_new_sessionstart_lock) APRMutex(mp);
      pipearray[i].localmp = NULL;
   }
   session_threads = (apr_thread_t **)apr_palloc(mp, sizeof(apr_thread_t *)*noofthreads);
   session_threadattrs = (apr_threadattr_t **)apr_palloc(mp, sizeof(apr_threadattr_t *)*noofthreads);
}

/** Destructor for the basic TCP server. */
TCPServer::~TCPServer()
{
   joinThreads();
   // don't do free on pipearray, session_threads
   apr_pool_clear(mp);
   apr_pool_destroy(mp);
}

/** Start the listening thread. This must be started first. */
int TCPServer::startListeningThread()
{
   apr_status_t status;
   status = apr_threadattr_create(&listening_attr, mp);
   CheckAPRError(status);
   status = apr_thread_create(&listening_thread, listening_attr, (functor)(&(TCPServer::handleServerListeningStatic)), (void *)this, mp);
   CheckAPRError(status);
   return 0;
}

/** Start the session threads. These must be started after the
 * listening thread. */
int TCPServer::startSessionThreads()
{
   void ***tempvoid;
   tempvoid = (void ***)apr_palloc(mp, sizeof(void **)*noofthreads);
   apr_status_t status;
   for (int i = 0; i < noofthreads; i++)
   {
      tempvoid[i] = (void **)apr_palloc(mp, sizeof(void *)*2);
      tempvoid[i][0] = (void *)apr_palloc(mp, sizeof(void *));
      tempvoid[i][1] = (void *)apr_palloc(mp, sizeof(int));
      *((void **)tempvoid[i][0]) = (void *)this;
      *((int *)tempvoid[i][1]) = i;
      status = apr_threadattr_create(&(session_threadattrs[i]), mp);
      CheckAPRError(status);
      // Create stack protectors to prevent stack smashing attacks
      //status = apr_threadattr_guardsize_set(session_threadattrs[i],
      //4096); CheckAPRError(status);
      status = apr_thread_create(&(session_threads[i]), session_threadattrs[i], (functor)(getClientHandlerFunction()), (void *)tempvoid[i], mp);
      CheckAPRError(status);
   }
   return 0;
}

/** Accept the incoming connection on the socket.
 * 
 * @param pollset Pointer to the set of existing connections to be
 * polled for data.
 * 
 * @param lsock The socket pointer.
 * 
 * @param t_mp Memory pool from which to allocate new client handler
 * pipes. */
void TCPServer::acceptIncomingOnSocket(apr_pollset_t *pollset, apr_socket_t *lsock, apr_pool_t *t_mp)
{
   // Must find inactive socket, or this does nothing (which is good!)
   apr_status_t status;
   for (int j = 0; j < noofthreads; j++)
   {
      if ((pipearray[j].recv_to_handler == NULL) && (pipearray[j].handler_to_send == NULL))
      {
         pipearray[j].sessionstart->lock();
         status = apr_pool_create(&(pipearray[j].localmp), t_mp);
         CheckAPRError(status);
         void *mem_for_recv_to_handler = (APRBucketBrigade *)apr_palloc(t_mp, sizeof(APRBucketBrigade));
         void *mem_for_handler_to_send = (APRBucketBrigade *)apr_palloc(t_mp, sizeof(APRBucketBrigade));
         pipearray[j].recv_to_handler = new(mem_for_recv_to_handler) APRBucketBrigade(t_mp);
         pipearray[j].handler_to_send = new(mem_for_handler_to_send) APRBucketBrigade(t_mp);
         status = apr_socket_accept(&(pipearray[j].clisock), lsock, t_mp);
         CheckAPRError(status);
         status = apr_socket_addr_get(&(pipearray[j].cliaddr), APR_REMOTE, pipearray[j].clisock);
         CheckAPRError(status);
         apr_pollfd_t pfd = { pipearray[j].localmp, APR_POLL_SOCKET, APR_POLLIN|APR_POLLOUT|APR_POLLHUP|APR_POLLERR|APR_POLLNVAL, 0, { NULL }, NULL };
         pfd.desc.s = pipearray[j].clisock;
         status = apr_socket_opt_set(pipearray[j].clisock, APR_SO_NONBLOCK, 1);
         CheckAPRError(status);
         status = apr_socket_timeout_set(pipearray[j].clisock, DATA_TIMEOUT);
         CheckAPRError(status);
         status = apr_pollset_add(pollset, &pfd);
         CheckAPRError(status);
         pipearray[j].sessionstart->signal();
         pipearray[j].sessionstart->unlock();
         return;
      }
   }
}

/** Start listening on the socket (the listening thread is this
 * function).
 * 
 * @param localmp The local memory pool to this listening function. */
void TCPServer::handleServerListening(apr_pool_t *localmp)
{
   apr_status_t status;
   apr_socket_t *lsock = createListeningSocket(localmp);
   if (lsock == NULL) { listenstop = 1; return; }
   apr_pollset_t *pollset;
   status = apr_pollset_create(&pollset, noofthreads, localmp, 0);
   CheckAPRError(status);
   apr_pollfd_t mainpollfd = { localmp, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, NULL };
   const apr_pollfd_t *workingpollfd;
   mainpollfd.desc.s = lsock;
   status = apr_pollset_add(pollset, &mainpollfd);
   CheckAPRError(status);
   char *data = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
   while (!listenstop)
   {
      apr_int32_t fdnum;
      status = apr_pollset_poll(pollset, POLL_TIMEOUT_USEC, &fdnum, &workingpollfd);
      if (!(APR_STATUS_IS_EINTR(status) || APR_STATUS_IS_TIMEUP(status)))
      {
         CheckAPRError(status);
         //if (APR_STATUS_IS_EBADF(status))
      }
      int i = -1;
      for (i = 0; i < fdnum; i++)
      {
         int handled = 0;
         if (workingpollfd[i].desc.s == lsock)
         {
            acceptIncomingOnSocket(pollset, lsock, localmp);
            handled = 1;
         } else
            for (int j = 0; j < noofthreads; j++)
               if ((pipearray[j].recv_to_handler) && (pipearray[j].handler_to_send))
                  if (pipearray[j].clisock == workingpollfd[i].desc.s)
                  {
                     if (workingpollfd[i].rtnevents & APR_POLLIN)
                        if (pipearray[j].recv_to_handler->packedSpaceRemaining())
                        {
                           pipearray[j].waiting->lock();
                           apr_size_t dataquant = MAX_BLOCK_SIZE;
                           status = apr_socket_recv(pipearray[j].clisock, data, &dataquant);
                           if (status != APR_SUCCESS)
                           {
                              pipearray[j].recv_to_handler->writeEOS();
                              if (!(APR_STATUS_IS_EOF(status)))
                              {
                                 CheckAPRError(status);
                                 if (APR_STATUS_IS_EBADF(status))
                                    pipearray[j].handler_to_send->writeEOS();
                              }
                           } else if (dataquant > 0)
                              pipearray[j].recv_to_handler->writeToStream(data, dataquant);
                           pipearray[j].waiting->signal();
                           pipearray[j].waiting->unlock();
                        }
                     if (workingpollfd[i].rtnevents & APR_POLLOUT)
                        if (pipearray[j].handler_to_send->hasData())
                        {
                           pipearray[j].waiting->lock();
                           apr_size_t dataquant = pipearray[j].handler_to_send->readFromStream(data, MAX_BLOCK_SIZE);
                           status = apr_socket_send(pipearray[j].clisock, data, &dataquant);
                           if (status != APR_SUCCESS)
                           {
                              pipearray[j].handler_to_send->clear();
                              pipearray[j].handler_to_send->writeEOS();
                              CheckAPRError(status);
                           } else if (dataquant > 0)
                              pipearray[j].handler_to_send->popFromStream(data, dataquant);
                           pipearray[j].waiting->signal();
                           pipearray[j].waiting->unlock();
                        }
                     if (workingpollfd[i].rtnevents & (APR_POLLHUP|APR_POLLERR|APR_POLLNVAL))
                     {
                        pipearray[j].waiting->lock();
                        pipearray[j].recv_to_handler->writeEOS();
                        pipearray[j].handler_to_send->clear();
                        pipearray[j].handler_to_send->writeEOS();
                        pipearray[j].waiting->signal();
                        pipearray[j].waiting->unlock();
                     }
                     handled = 1;
                  }
         if (handled == 0)
            errprint("Unmatched/unknown file descriptor in poll.");
      }
      for (int j = 0; j < noofthreads; j++)
         if ((pipearray[j].recv_to_handler) && (pipearray[j].handler_to_send))
            if ((pipearray[j].recv_to_handler->isClosed()) && (pipearray[j].handler_to_send->isClosed()))
            {
               pipearray[j].waiting->lock();
               apr_pollfd_t pfd = { pipearray[j].localmp, APR_POLL_SOCKET, APR_POLLIN|APR_POLLOUT|APR_POLLHUP|APR_POLLERR|APR_POLLNVAL, 0, { NULL }, NULL };
               pfd.desc.s = pipearray[j].clisock;
               status = apr_pollset_remove(pollset, &pfd);
               CheckAPRError(status);
               status = apr_socket_close(pipearray[j].clisock);
               CheckAPRError(status);
               // Free bucket brigades
               if (pipearray[j].recv_to_handler)
               {
                  APRBucketBrigade *pntr_brge = pipearray[j].recv_to_handler;
                  pipearray[j].recv_to_handler = NULL;
                  pntr_brge->~APRBucketBrigade();
               }
               if (pipearray[j].handler_to_send)
               {
                  APRBucketBrigade *pntr_brge = pipearray[j].handler_to_send;
                  pipearray[j].handler_to_send = NULL;
                  pntr_brge->~APRBucketBrigade();
               }
               apr_pool_clear(pipearray[j].localmp);
               apr_pool_destroy(pipearray[j].localmp);
               pipearray[j].localmp = NULL;
               pipearray[j].waiting->signal();
               pipearray[j].waiting->unlock();
            }
   }
   free(data);
   listenstop = 1;
   // Wasn't here, don't know whether this helps or hinders?!?
   apr_socket_close(lsock);
}

/** Start listening on the socket (the listening thread is this
 * function).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param vthis The this pointer carried into the thread.
 * 
 * @return Return is NULL. */
void *TCPServer::handleServerListeningStatic(apr_thread_t *tthread, void *vthis)
{
   apr_status_t status;
   TCPServer *kthis = (TCPServer *)vthis;
   apr_pool_t *upperpool, *t_mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&t_mp, upperpool);
   CheckAPRError(status);
   
   kthis->handleServerListening(t_mp);

   // do not clear memory pool t_mp, as bucket brigades may be remaining
   return NULL;
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param threadid The integer thread ID number.
 * 
 * @param localmp The memory pool local to this thread. */
void TCPServer::handleClientsOnThread(int threadid, apr_pool_t *localmp)
{
   pipearray[threadid].sessionstart->lock();
   while (!sessionstop)
   {
      pipearray[threadid].sessionstart->wait();
      if (sessionstop) break;
      apr_pool_t *sp;
      apr_pool_create(&sp, localmp);
      pipearray[threadid].waiting->lock();
      while ((!(pipearray[threadid].handler_to_send->isClosed())) && 
             (!(pipearray[threadid].recv_to_handler->isClosed())))
      {
         if (!(pipearray[threadid].recv_to_handler->hasData()))
            pipearray[threadid].waiting->wait();
         if ((!(pipearray[threadid].handler_to_send->isClosed())) && 
             (!(pipearray[threadid].recv_to_handler->isClosed())))
         {
            char *buffer = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
            if (pipearray[threadid].recv_to_handler->isClosed())
               pipearray[threadid].handler_to_send->writeEOS();
            while ((pipearray[threadid].handler_to_send->packedSpaceRemaining() >= MAX_BLOCK_SIZE) && 
                   (pipearray[threadid].recv_to_handler->hasData()))
            {
               apr_size_t dataquant = pipearray[threadid].recv_to_handler->popFromStream(buffer, MAX_BLOCK_SIZE);
               apr_size_t dataquant2 = pipearray[threadid].handler_to_send->writeToStream(buffer, dataquant);
            }
            free(buffer);
         }
      }
      // Make sure these are closed so that this pipe can be reused.
      pipearray[threadid].recv_to_handler->clear();
      pipearray[threadid].recv_to_handler->writeEOS();
      pipearray[threadid].handler_to_send->clear();
      pipearray[threadid].handler_to_send->writeEOS();

      pipearray[threadid].waiting->unlock();
      apr_pool_clear(sp);
      apr_pool_destroy(sp);
   }
   pipearray[threadid].sessionstart->unlock();
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *TCPServer::handleClientsOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   TCPServer *kthis = *((TCPServer **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *t_mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&t_mp, upperpool);
   CheckAPRError(status);

   kthis->handleClientsOnThread(i, t_mp);

   // do not clear t_mp, as bucket brigades may be remaining
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Join threads to main thread (stop all sessions). */
int TCPServer::joinThreads()
{
   apr_status_t status, thread_status;

   listenstop = 1;
   status = apr_thread_join(&thread_status, listening_thread);
   CheckAPRError(status);

   sessionstop = 1;
   for (int i = 0; i < noofthreads; i++)
   {
      // Finish this session
      if (pipearray[i].waiting)
      {
         pipearray[i].waiting->lock();
         // Clear and close streams, if they exist.
         if (pipearray[i].recv_to_handler)
         {
            pipearray[i].recv_to_handler->clear();
            pipearray[i].recv_to_handler->writeEOS();
         }
         if (pipearray[i].handler_to_send)
         {
            pipearray[i].handler_to_send->clear();
            pipearray[i].handler_to_send->writeEOS();
         }
         pipearray[i].waiting->signal();
         pipearray[i].waiting->unlock();
      }
      // Finish all sessions
      if (pipearray[i].sessionstart)
      {
         pipearray[i].sessionstart->lock();
         // Free bucket brigades
         if (pipearray[i].recv_to_handler)
         {
            APRBucketBrigade *pntr_brge = pipearray[i].recv_to_handler;
            pipearray[i].recv_to_handler = NULL;
            pntr_brge->~APRBucketBrigade();
         }
         if (pipearray[i].handler_to_send)
         {
            APRBucketBrigade *pntr_brge = pipearray[i].handler_to_send;
            pipearray[i].handler_to_send = NULL;
            pntr_brge->~APRBucketBrigade();
         }
         pipearray[i].sessionstart->signal();
         pipearray[i].sessionstart->unlock();
      }
      status = apr_thread_join(&thread_status, session_threads[i]);
      CheckAPRError(status);
   }

   return 0;
}

/** Make a listening socket from the memory pool
 * 
 * @param t_mp Memory pool to use.
 * 
 * @return Pointer to new listening socket. */
apr_socket_t *TCPServer::createListeningSocket(apr_pool_t *t_mp)
{
   apr_status_t status;
   apr_socket_t *tsocket;
   apr_sockaddr_t *addrstruct;

   // NULL here gives the 0.0.0.0 that is binding to all network
   // interfaces
   status = apr_sockaddr_info_get(&addrstruct, hostname, APR_UNSPEC, portno, 0, t_mp);
   CheckAPRError(status);
   status = apr_socket_create(&tsocket, addrstruct->family, SOCK_STREAM, APR_PROTO_TCP, t_mp);
   CheckAPRError(status);
   status = apr_socket_opt_set(tsocket, APR_SO_NONBLOCK, 1);
   CheckAPRError(status);
   status = apr_socket_timeout_set(tsocket, CONNECTION_TIMEOUT);
   CheckAPRError(status);
   status = apr_socket_opt_set(tsocket, APR_SO_REUSEADDR, 1);
   CheckAPRError(status);
   status = apr_socket_bind(tsocket, addrstruct);
   CheckAPRError(status);
   if (status != APR_SUCCESS)
   {
      errprint("bind to '%s:%d' failed", hostname, portno);
      return NULL;
   }
   // makes the number of backlog connections the most the OS will
   // support
   status = apr_socket_listen(tsocket, SOMAXCONN);
   CheckAPRError(status);

   listenstop = 0;
   return tsocket;
}

/** Get the static client address string from the client address
 * pointer in the pipearray. The returned pointer is only valid for as
 * long as the socket remains open.
 * 
 * @param threadid The ID of the thread who client address is to be
 * obtained.
 * 
 * @return Pointer to the temporarily available string. */
const char *TCPServer::getClientAddressAsStaticString(int threadid)
{
   apr_status_t status;
   char *ipaddress;
   status = apr_sockaddr_ip_get(&ipaddress, pipearray[threadid].cliaddr);
   CheckAPRError(status);
   return ipaddress;
}
